package com.abyss.source;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.util.Random;
import java.util.UUID;

/**
 * 构建一个可以并行的自定义Source
 */
public class CustomParallelSourceDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<OrderInfo2> source = executionEnvironment.addSource(new MySourceWithParallel()).setParallelism(6);

        source.print();

        executionEnvironment.execute();
    }

    public static class MySourceWithParallel implements ParallelSourceFunction<OrderInfo2> {
        private boolean isRun = true;
        @Override
        public void run(SourceContext<OrderInfo2> ctx) throws Exception {
            Random random = new Random();
            while (this.isRun) {
                String orderId = UUID.randomUUID().toString();
                String userId = UUID.randomUUID().toString();
                int money = random.nextInt(99);
                long ts = System.currentTimeMillis();
                // 通过ctx对象的collect方法将要提供的数据返回出去
                ctx.collect(new OrderInfo2(orderId, userId, money, ts));
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            this.isRun = false;
        }
    }


    public static class OrderInfo2 {
        private String orderId;
        private String userId;
        private int money;
        private long ts;

        @Override
        public String toString() {
            return "OrderInfo{" +
                    "orderId='" + orderId + '\'' +
                    ", userId='" + userId + '\'' +
                    ", money=" + money +
                    ", ts=" + ts +
                    '}';
        }

        public String getOrderId() {
            return orderId;
        }

        public void setOrderId(String orderId) {
            this.orderId = orderId;
        }

        public String getUserId() {
            return userId;
        }

        public void setUserId(String userId) {
            this.userId = userId;
        }

        public int getMoney() {
            return money;
        }

        public void setMoney(int money) {
            this.money = money;
        }

        public long getTs() {
            return ts;
        }

        public void setTs(long ts) {
            this.ts = ts;
        }

        public OrderInfo2() {
        }

        public OrderInfo2(String orderId, String userId, int money, long ts) {
            this.orderId = orderId;
            this.userId = userId;
            this.money = money;
            this.ts = ts;
        }
    }
}